package com.atguigu.gmall.realtime.utils;


import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;

import java.util.Properties;

/*
操作Kafka的工具类
 */
public class MyKafkaUtil {
    private static String KafkaServer="localhost:9092,localhost:9093,localhost:9094";
    private static final String DEFAULT_TOPIC="DEFAULT_DATA";

    public static FlinkKafkaConsumer<String> getKafkaSource(String topic, String groupId){
        Properties props = new Properties();
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG,groupId);
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaServer);
        return new FlinkKafkaConsumer<String>(topic,new SimpleStringSchema(),props);
    }

    public static FlinkKafkaConsumer<String> getKafkaSource(String testKafkaServer,String topic, String groupId){
        Properties props = new Properties();
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG,groupId);
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, testKafkaServer);
        return new FlinkKafkaConsumer<String>(topic,new SimpleStringSchema(),props);
    }


    public static FlinkKafkaProducer<String> getKafkaSink(String topic){
        return new FlinkKafkaProducer<String>(KafkaServer,topic,new SimpleStringSchema());
    }

    //SimpleSchema 做序列化操作使用
    public static <T> FlinkKafkaProducer<T> getKafkaSinkBySchema(KafkaSerializationSchema<T> serializationSchema){
        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,KafkaServer);
        properties.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,1000*60*15+"");
        return new FlinkKafkaProducer<T>(DEFAULT_TOPIC,serializationSchema,properties,
                FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
    }


    //拼接kafka相关属性到DDL
    public static String getKafkaDDL(String topic,String groupId){
        String ddl = "'connector' = 'kafka', "+
                "'topic' = '"+topic+"'," +
                "'properties.bootstrap.servers' = '" + KafkaServer +"', " +
                "'properties.group.id' = '"+groupId + "', " +
                "'format' ='json'," +
                "'scan.startup.mode' = 'latest-offset'  ";
        return ddl;
    }


    //拼接kafka相关属性到DDL
    public static String getKafkaDDLTest(String topic,String groupId,String kafka){
        String ddl = "'connector' = 'kafka', "+
                "'topic' = '"+topic+"'," +
                "'properties.bootstrap.servers' = '" + kafka +"', " +
                "'properties.group.id' = '"+groupId + "', " +
                "'format' ='csv'," +
                "'scan.startup.mode' = 'latest-offset' ";
        return ddl;
    }
}
